package util

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import util.Eastmoney.StockData

import scala.util.Random


class RandDomData extends SourceFunction[StockData] {
  var running = true

  override def run(ctx: SourceContext[StockData]): Unit = {
    val random = new Random

    // 预定义100个不同的股票代码和公司名称，以及相关链接
    val codes = (1 to 100).map(i => f"SH60$i%04d").toArray
    val names = Array.tabulate(100)(i => s"Company_${i + 1}")
    val links = Array("股吧 资金流 数据")

    while (running) {
      // 获取当前时间的时间戳
      val currentTimeMillis = System.currentTimeMillis()

      // 随机选择索引、代码、名称和链接
      val index = random.nextInt(100) + 1 // 索引编号从1到100
      val code = codes(random.nextInt(codes.length))
      val name = names(random.nextInt(names.length))
      val link = links(random.nextInt(links.length))

      // 使用随机值模拟变动的数据点
      val latestPrice = 8.5 + random.nextDouble() * 1.0 // 在8.5到9.5之间随机
      val changePercentage = f"${-3.0 + random.nextDouble() * 2.0}%.2f%%" // 在-3%到-1%之间随机
      val changeAmount = -0.3 + random.nextDouble() * 0.2 // 在-0.3到-0.1之间随机
      val volumeHands = 600000L + random.nextLong().abs % 300000L // 在600,000和900,000之间随机
      val turnoverYuan = "56709.61亿" // 这个值似乎是固定的
      val amplitude = f"${3.0 + random.nextDouble() * 2.0}%.2f%%" // 在3%到5%之间随机
      val high = 3.5 + random.nextDouble() * 2.0 // 在3.5到5.5之间随机
      val low = 8.0 + random.nextDouble() * 2.0 // 修改了范围：在8.0到10.0之间随机，更合理
      val openPrice = 8.5 + random.nextDouble() * 1.0 // 在8.5到9.5之间随机，与最新价格范围一致
      val previousClose = latestPrice - changeAmount
      val turnoverRate = f"${1.0 + random.nextDouble() * 0.5}%.2f%%" // 在1%到1.5%之间随机
      val peRatioDynamic = 35.0 + random.nextDouble() * 5.0 // 在35到40之间随机
      val pbRatio = 10.0 + random.nextDouble() * 2.0 // 在10到12之间随机

      // 发送构造的数据对象，并设置 writeTime 和 updateTime 为当前时间戳
      ctx.collect(StockData(
        index,
        code,
        name,
        link,
        latestPrice,
        changePercentage,
        changeAmount,
        volumeHands,
        turnoverYuan,
        amplitude,
        high,
        low,
        openPrice,
        previousClose,
        turnoverRate,
        peRatioDynamic,
        pbRatio,
        writeTime = currentTimeMillis,
        updateTime = currentTimeMillis
      ))

      Thread.sleep(1000) // 每隔一秒发送一次数据
    }
  }

  override def cancel(): Unit = running = false
}